package ex55.mm.flow.impound.quote.exchange.api;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class WebSocketWatchDog {

    private final CopyOnWriteArrayList<MarketDepthWSConnection> TIME_HELPER =
            new CopyOnWriteArrayList<>();

    private final SubscriptionOptions options;

    private static final Logger log = LoggerFactory.getLogger(WebSocketWatchDog.class);

    public WebSocketWatchDog(SubscriptionOptions subscriptionOptions) {
        this.options = Objects.requireNonNull(subscriptionOptions);
        long t = 1_000;
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1);
        exec.scheduleAtFixedRate(
                () -> {
                    TIME_HELPER.forEach(
                            connection -> {
                                try {
                                    if (connection.getState()
                                            == MarketDepthWSConnection.ConnectionState.CONNECTED) {
                                        // Check response
                                        if (options.isAutoReconnect()) {
                                            long ts =
                                                    System.currentTimeMillis()
                                                            - connection.getLastReceivedTime();
                                            if (ts > options.getReceiveLimitMs()) {
                                                log.warn("MarketDepthConnection({}.{}) No response from server", connection.getExchangeId(), connection.getProduct().getSymbol());
                                                connection.reConnect(
                                                        options.getConnectionDelayOnFailure());
                                            }
                                        }
                                    } else if (connection.getState()
                                            == MarketDepthWSConnection.ConnectionState.DELAY_CONNECT) {
                                        connection.reConnect();
                                    } else if (connection.getState()
                                            == MarketDepthWSConnection.ConnectionState.CLOSED_ON_ERROR) {
                                        if (options.isAutoReconnect()) {
                                            connection.reConnect(options.getConnectionDelayOnFailure());
                                        }
                                    }
                                } catch (Exception ex) {
                                    ex.printStackTrace();
                                }

                            });
                },
                t,
                t,
                TimeUnit.MILLISECONDS);
        Runtime.getRuntime().addShutdownHook(new Thread(exec::shutdown));
    }

    void onConnectionCreated(MarketDepthWSConnection connection) {
        TIME_HELPER.addIfAbsent(connection);
    }

    void onClosedNormally(MarketDepthWSConnection connection) {
        TIME_HELPER.remove(connection);
    }

    public SubscriptionOptions getOptions() {
        return options;
    }
}
